package com.xs.mvc.kafka;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.MessageListener;

import com.alibaba.fastjson.JSON;

/**
 * @Title: <P>KafkaSendMessageServiceImpl.java</P>
 * @Package <P>com.xs.mvc.kafka</P>
 * @Description: <P>TODO该类...</P>
 * @author renweiping
 * @date 2019年3月12日 上午9:35:41
 * @version 1.0
 */

public class KafkaSendMessageService implements MessageListener<String,String>{

	protected final Log logger = LogFactory.getLog(this.getClass()); // NOSONAR
	/* (non-Javadoc)
	 * @see org.springframework.kafka.listener.MessageListener#onMessage(org.apache.kafka.clients.consumer.ConsumerRecord)
	 */
	@Override
	public void onMessage(ConsumerRecord<String, String> data) {
		 //根据不同主题，消费
       /* System.out.println("========");
        if("topic1".equals(data.topic())){
            //逻辑1
            System.out.println(data.value()+"被消费");
        }else if("topic2".equals(data.topic())){
            //逻辑2
            System.out.println(data.value()+"主题2 被消费");
        }else {
        	System.out.println("消费者收到数据为：" + JSON.toJSONString(data));
        	
        	
        }*/
        logger.info("thread is "+Thread.currentThread().getName()+"||***************消费者收到数据为：{\"topic\":" +data.topic()+",\"partition\"="+data.partition()+",\"offset\"="+data.offset()
        		+",\"timestamp\"="+data.timestamp()+",\"value\"="+data.value()+",\"key\"="+data.key()+"}");
        		
	}

}
